package com.ustcinfo.ishare.kafka.monitor.controller;

import com.ustcinfo.ishare.kafka.monitor.bean.BrokerInfo;
import com.ustcinfo.ishare.kafka.monitor.bean.KafkaConfigFile;
import com.ustcinfo.ishare.kafka.monitor.bean.KafkaMonitorInfo;
import com.ustcinfo.ishare.kafka.monitor.bean.MetadataInfo;
import com.ustcinfo.ishare.kafka.monitor.service.IKafkaConfigFileService;
import com.ustcinfo.ishare.kafka.monitor.utils.GetAllKafkaInfo;
import com.ustcinfo.ishare.kafka.monitor.service.IKafkaInfoService;
import com.ustcinfo.ishare.kafka.monitor.utils.KafkaInfoUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.servlet.http.HttpServletRequest;
import java.io.File;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by Shinelon on 2018/3/26.
 */
@Controller
@RequestMapping("/getKafkaMonitorInfo")
public class GetKafkaMonitorInfo {

    @Autowired
    IKafkaInfoService kafkaInfoService;
    @Autowired
    KafkaInfoUtils kafkaInfoUtils;
    @Autowired
    GetAllKafkaInfo getAllKafkaInfo ;
    @Autowired
    IKafkaConfigFileService kafkaConfigFileService;

    @RequestMapping("/getAllMonitorInfo")
    public String getAllMonitorInfo(Model model) {
        Map<String, KafkaMonitorInfo> kafkaMonitorInfoMap = GetAllKafkaInfo.getLatestMinuteMap();

        List<KafkaMonitorInfo> kafkaMonitorInfoList = new ArrayList<KafkaMonitorInfo>();
        for (Map.Entry<String, KafkaMonitorInfo> entry : kafkaMonitorInfoMap.entrySet()) {
            kafkaMonitorInfoList.add(entry.getValue());
        }

        model.addAttribute("listData", kafkaMonitorInfoList);
        return "topicInfo";
    }

    @RequestMapping(value = "/ajaxGetKafkaMonitorInfo")
    @ResponseBody
    public Map<String,List<KafkaMonitorInfo>> ajaxGetKafkaMonitorInfo(){
        List<KafkaMonitorInfo> kafkaMonitorInfoList = new ArrayList<KafkaMonitorInfo>();

        Map<String, KafkaMonitorInfo> kafkaMonitorInfoMap = GetAllKafkaInfo.getLatestMinuteMap();
        for (Map.Entry<String, KafkaMonitorInfo> entry : kafkaMonitorInfoMap.entrySet()) {
            kafkaMonitorInfoList.add(entry.getValue());
        }

        Map<String, List<KafkaMonitorInfo>> monitorInfoMap = new HashMap<String, List<KafkaMonitorInfo>>();
        monitorInfoMap.put("listData",kafkaMonitorInfoList);
        return monitorInfoMap;
    }

    @RequestMapping("/getBrokerInfo")
    public String getBrokerInfo(Model model) {
        Set<BrokerInfo> brokerInfos = kafkaInfoUtils.getAllBrokersInfo();
        model.addAttribute("brokerData", brokerInfos);
        return "brokerInfo";
    }

    @RequestMapping("/getTopicMetadataInfo")
    public String getTopicMetadataInfo(Model model) {
        List<MetadataInfo> metadataInfoList = getAllKafkaInfo.getAllMetadataInfo();
        model.addAttribute("topicMetadata",metadataInfoList);
        return "topicMetadataInfo";
    }

    @RequestMapping("/searchTopic")
    public String searchTopic(Model model) {
        model.addAttribute("listData","");
        return "topicInfoSearch";
    }

    @RequestMapping(value = "/searchTopicFromMysql")
    public String searchTopicFromMysql(HttpServletRequest request ,Model model){
        String topicName = request.getParameter("topicName");
        String groupName = request.getParameter("groupName");
        String beginTime = request.getParameter("beginTime");
        String endTime = request.getParameter("endTime");

        Long beginTime3 = 0L;
        Long endTime3 = 2000000000000l;

        try {
            if(!beginTime.equals("")){
                Date beginTime2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(beginTime);
                beginTime3 = beginTime2.getTime();
            }
            if(!endTime.equals("")){
                Date endTime2 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(beginTime);
                endTime3 = endTime2.getTime();
            }
        } catch (ParseException e) {
            e.printStackTrace();
        }

        List<KafkaMonitorInfo> kafkaMonitorInfos = kafkaInfoService.findKafkaMonitorInfo(topicName,groupName,beginTime3,endTime3);

        model.addAttribute("listData", kafkaMonitorInfos);
        return "topicInfoSearch";
    }

    @RequestMapping("/getTopicList")
    public String getTopicList(Model model) {
        List<MetadataInfo> topicList = getAllKafkaInfo.getTopicList();
        model.addAttribute("topicList",topicList);
        return "topicList";
    }

    @RequestMapping("/createTopic")
    public String createTopic() {
        return "createTopic";
    }

    @RequestMapping("/ajaxCreateTopic")
    @ResponseBody
    public String ajaxCreateTopic(HttpServletRequest request) {
        String topicName = request.getParameter("topicName");
        String partition = request.getParameter("partition");
        String replication = request.getParameter("replication");
        String s = kafkaInfoUtils.createTopic(topicName,partition,replication);
        return s;
    }

    @RequestMapping("/ajaxDeleteTopic")
    @ResponseBody
    public String ajaxDeleteTopic(HttpServletRequest request) {
        String topicName = request.getParameter("topicName");
        String s = kafkaInfoUtils.deleteTopic(topicName);
        return s;
    }

    @RequestMapping(value = "/existsTopic")
    @ResponseBody
    public String existsTopic(HttpServletRequest request){
        String topicName = request.getParameter("topicName");
        List<String> topics = kafkaInfoUtils.getAllTopics();
        return topics.contains(topicName)+"";
    }

    @RequestMapping("/getConfigFileList")
    public String getConfigFileList(Model model) {
        List<KafkaConfigFile> kafkaConfigFileList = kafkaConfigFileService.findAllKafkaConfigFile();
        model.addAttribute("kafkaConfigFileList",kafkaConfigFileList);
        return "configFileManager";
    }

    @RequestMapping(value = "/deleteConfigFile")
    @ResponseBody
    public String deleteConfigFile(HttpServletRequest request){
        Integer id = Integer.valueOf(request.getParameter("id"));
        KafkaConfigFile kafkaConfigFile = kafkaConfigFileService.getKafkaConfigFileById(id);

        kafkaConfigFileService.removeById(id);

        File file = new File(kafkaConfigFile.getConfigFileUrl());
        file.delete();
        return "true";
    }

}
